package com.migo.utils;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

public class KafkaUtil {
    public static final String GROUPID = "AppLogGroup";

    public static Producer<String, String> createProducer() {
        Properties props = new Properties();
        // 设置group id
        props.put("group.id", GROUPID);
//        props.put("zookeeper.connect","123.57.163.139:2188");
        props.put("metadata.broker.list", "123.57.163.139:9092");
//        props.put("metadata.broker.list", "123.57.163.139:9092,114.215.28.4:9092,47.89.31.122:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("request.required.acks", "1");
        props.put("", "");
        ProducerConfig config = new ProducerConfig(props);
        return new Producer<String, String>(config);
    }

    public static void sendData(Producer<String, String> producer, String topic, String logId, String log) {
        KeyedMessage<String, String> data = new KeyedMessage<>(topic, logId, log);
        producer.send(data);
        producer.close();
    }
}
